#include <stdlib.h>
#include <errno.h>
#include <pthread.h>
#include <poll.h>
#include <signal.h>
#include <netinet/tcp.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>

#include "connpool.h"
#include "error_code.h"
#include "fcgi_stdio.h"
#include "base.h"

#define CMD_DUMMY	255

void
signal_mask(int sig)
{
	sigset_t sig_set;
	sigemptyset(&sig_set);
	sigaddset(&sig_set, sig);
	sigprocmask(SIG_BLOCK, &sig_set, NULL);
}

void
signal_unmask(int sig)
{
	sigset_t sig_set;
	sigemptyset(&sig_set);
	sigaddset(&sig_set, sig);
	sigprocmask(SIG_UNBLOCK, &sig_set, NULL);
}

static struct reactive_queue_head *reactive_q;
static ConnPool *cp;

/*
int memlink_cmd_dummy(MemLink *m)
{
	char buf[1024];
	int len = pack(buf, 0, "$4c", CMD_DUMMY);
	int ret;
	if (m->writefd <= 0) {
		int fd = socket(AF_INET, SOCK_STREAM, 0);
		if (fd == -1) {
			DERROR("socket error\n", strerror(errno));
			return -1;
		}
		m->writefd = fd;
		struct linger ling = {1, 0};
		setsockopt(fd, SOL_SOCKET, SO_LINGER, &ling, sizeof(ling));
		int f = 1;
		setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &f, sizeof(f));
		struct sockaddr_in s;
		s.sin_family = AF_INET;
		s.sin_port = htons(m->write_port);
		inet_pton(AF_INET, m->host, &s.sin_addr);
		ret = connect(fd, (struct sockaddr *)&s, sizeof(s));
		if (ret == -1) {
			DERROR("connect to memlink error: %s\n", strerror(errno));
			return -1;
		}
	}

	char *p = buf;
	int left = len;
	int nwrite = 0;
	while (left > 0) {
		ret = write(m->writefd, p+nwrite, left);
		if (ret < 0) {
			if (errno == EINTR)
				continue;
			return -1;
		}
		nwrite += ret;
		left -= ret;
	}

	int nread = read(m->writefd, buf, 1024);
	if (nread > 0)
		return MEMLINK_OK;
	else
		return -1;
}
*/
ConnPool *init_connpool(const Config* conf)
{
	ConnPool *pool = new ConnPool;
	cp = pool;
	int ret = pool->init(conf);

	if (ret == -1)
    {
        delete pool;
        return NULL;
    }

	return pool;
}

void close_connpool(ConnPool *pool)
{
	delete(pool);
}

ConnPoolItem *queue_get()
{
	ConnPoolItem *conn = NULL;
	struct reactive_queue_item *item;

	pthread_mutex_lock(&reactive_q->m);
	DERROR("queue num: %d\n", reactive_q->num);
	if (reactive_q->root) {
		item = reactive_q->root;
		reactive_q->root = item->next;
		if (reactive_q->root == NULL)
			reactive_q->tail = NULL;
		conn = item->conn;
		delete(item);
	}
	reactive_q->num--;
	pthread_mutex_unlock(&reactive_q->m);

	return conn;
}

int queue_put(ConnPoolItem *conn)
{
	struct reactive_queue_item *item = new(struct reactive_queue_item);
	item->conn = conn;
	item->next = NULL;

	pthread_mutex_lock(&reactive_q->m);
	if (reactive_q->root == NULL) {
		reactive_q->root = item;
	} else {
		reactive_q->tail->next = item;
	}
	reactive_q->tail = item;
	reactive_q->num++;
	pthread_mutex_unlock(&reactive_q->m);

	return 0;
}

struct reactive_queue_head *reactive_queue_init()
{
	struct reactive_queue_head *q = new(struct reactive_queue_head);
	pthread_mutex_init(&q->m, NULL);
	q->root = NULL;
	q->tail = NULL;
	q->num = 0;
	return q;
}

void *monitor(void *arg)
{
	struct pollfd pfd[2];
	char buf[1024];

	signal_mask(SIGTERM);
	memset(&pfd, 0, sizeof(pfd));

	pfd[0].fd = cp->fds->reactive_fd[0];
	pfd[0].events = POLLIN;

	pfd[1].fd = cp->fds->memlink_timer_fd[0];
	pfd[1].events = POLLIN;

	reactive_q = reactive_queue_init();
	DINFO("monitor start\n");

	while (1) {
        //DINFO("monitor running");
        int ret = poll(pfd, 2, -1);
		if (ret < 0) {
			if (errno == EINTR) {
				//DERROR("poll error: %s\n", strerror(errno));
				continue;
			} else {
				DERROR("poll error: %s\n", strerror(errno));
				abort();
			}
		} else if (ret == 0) {
			DERROR("monitor: %d\n", ret);
			continue;
		}
		//DERROR("accept notify\n");
		if ((pfd[0].revents & POLLIN) == POLLIN) {

			DINFO("one conn broken\n");
			ConnPoolItem *conn = queue_get();
			if (conn) {
				if (conn->type == CONN_POOL_TYPE_HBASE) {
					struct hbase_conn *hconn = (struct hbase_conn *)conn->item;
					ret = hconn->reactive();
					if (ret == 0) {
						conn->active = 1;
						conn->idle = 1;
					} else {
						queue_put(conn);
						continue;
					}
				} else if (conn->type == CONN_POOL_TYPE_HDFS) {
					//struct hdfs_conn *hconn = (struct hdfs_conn *)conn->item;
					//hconn->reactive();
				} else if (conn->type == CONN_POOL_TYPE_MEMLINK) {
/*					struct memlink_conn *mconn = (struct memlink_conn *)conn->item;
					ret = mconn->reactive();
					if (ret == 0) {
						conn->active = 1;
						conn->idle = 1;
					} else {
						queue_put(conn);
						continue;
					}
*/				}
			}

			read(pfd[0].fd, buf, 1);
		}
        sleep(10);
        /*
		if (pfd[1].revents & POLLIN == POLLIN) {
			DINFO("memlink conn checker run\n");
			read(pfd[1].fd, buf, 1024);

			struct ConnPoolItemHead *head = &cp->pool[CONN_POOL_TYPE_MEMLINK];
			struct ConnPoolItem *conn = head->root;
			int get;

			while (conn) {
				get = 0;
				pthread_mutex_lock(&head->m);
				if (conn->active && conn->idle) {
					conn->idle = 0;
					get = 1;
				}
				pthread_mutex_unlock(&head->m);
				if (get) {
					DINFO("test conn: %d\n", conn->id);
					struct memlink_conn *mconn = (struct memlink_conn *)conn->item;
					MemLink *m = mconn->conn;

					if ((memlink_cmd_ping(m) != MEMLINK_OK) ||
						memlink_cmd_dummy(m) != MEMLINK_OK) {
						DERROR("one memlink conn broken, move to reactive queue\n");
						conn->active = 0;
						queue_put(conn);
						write(cp->fds->reactive_fd[1], "", 1);
					} else {
						conn->idle = 1;
					}
				} else {
					DINFO("in use: %d\n", conn->id);
				}
				conn = conn->next;
			}
		}
        */
	}
}

void handler_sigalrm(int sig)
{
	write(cp->fds->memlink_timer_fd[1], "", 1);
	return;
}

hbase_conn::hbase_conn(const char *ip, unsigned short port)
{
	conn = NULL;
	pool = NULL;
    
    memcpy(hostip, ip, sizeof(hostip));
    this->port = port;
}

int hbase_conn::reactive()
{
	int ret;
	close();
	ret = open();

	if (ret < 0) {
		close();
		return -1;
	}
	return 0;
}

int hbase_conn::open()
{
	conn = new HbaseOp(hostip, port);
	return conn->open();
}

int hbase_conn::close()
{
	if (conn) {
		conn->close();
		delete(conn);
		conn = NULL;
		//pool = NULL;
	}
	return 0;
}

hdfs_conn::hdfs_conn(const char *ip, unsigned short port)
{
    conn = NULL;
    pool = NULL;
    memcpy(hostip, ip, sizeof(hostip));
    this->port = port;
}

int hdfs_conn::reactive()
{
	int ret;
	close();
	ret = open();
    
	if (ret < 0) {
		close();
		return -1;
	}
	return 0;    
}

int hdfs_conn::open()
{
	conn = new HdfsOp(hostip, port);
	return conn->init();
}

int hdfs_conn::close()
{
	if (conn) {
		conn->uninit();
		delete(conn);
		conn = NULL;
		//pool = NULL;
	}
	return 0;
}

/*
memlink_conn::memlink_conn(const char *host, unsigned short rport, unsigned short wport)
{
	conn = NULL;
	pool = NULL;
    memcpy(hostip, host, sizeof(hostip));
    this->rport = rport;
    this->wport = wport;
}

int memlink_conn::reactive()
{
	close();
	if (open() < 0)
		return -1;
	return 0;
}

int memlink_conn::open()
{
	conn = memlink_create(hostip, rport, wport, 0);
	if (conn == NULL) {
		return -1;
	} else {
		if (memlink_cmd_ping(conn) != MEMLINK_OK ||
			memlink_cmd_dummy(conn) != MEMLINK_OK) {
			close();
			return -1;
		} else {
			return 0;
		}
	}
}

int memlink_conn::close()
{
	if (conn) {
		memlink_destroy(conn);
		conn = NULL;
		//pool = NULL;
	}
	return 0;
}
*/
ConnPool::ConnPool()
{
	memset(pool, 0, sizeof(pool));

	return;
}

ConnPool::~ConnPool()
{
	int i;
	for (i = 0; i < CONN_POOL_TYPE_NUM; i++) {
		ConnPoolItem *conn = pool[i].root;
		ConnPoolItem *c;
		pthread_mutex_lock(&pool[i].m);
		if (pool[i].type == CONN_POOL_TYPE_HBASE) {
			while (conn) {
				struct hbase_conn *hconn = (struct hbase_conn *)conn->item;
				if (hconn) {
					hconn->close();
					delete(hconn);
				}
				c = conn;
				conn = conn->next;
				delete(c);
			}
		} else if (pool[i].type == CONN_POOL_TYPE_HDFS) {
			while (conn) {
				struct hdfs_conn *hconn = (struct hdfs_conn *)conn->item;;
				if (hconn) {
					hconn->close();
					delete(hconn);
				}
				c = conn;
				conn = conn->next;
				delete(c);
			}
		} else if (pool[i].type == CONN_POOL_TYPE_MEMLINK) {
/*			while (conn) {
				struct memlink_conn *mconn = (struct memlink_conn *)conn->item;
				if (mconn) {
					mconn->close();
					delete(mconn);
				}
				c = conn;
				conn = conn->next;
				delete(c);
			}
*/		}
		pthread_mutex_unlock(&pool[i].m);
	}

	return;
}



int ConnPool::start_connpool_monitor()
{
	pthread_attr_t attr;
	int ret;

	if ((ret = pthread_attr_init(&attr)) != 0) {
		DERROR("pthread_attr_init error: %s\n", strerror(ret));
		return -1;
	}

	if ((ret = pthread_attr_setscope(&attr, PTHREAD_SCOPE_SYSTEM)) != 0) {
		DERROR("pthread_attr_setscope error: %s\n", strerror(ret));
		pthread_attr_destroy(&attr);
		return -1;
	}

	if ((ret = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED)) != 0) {
		DERROR("pthread_attr_setdetachstate error: %s\n", strerror(ret));
		pthread_attr_destroy(&attr);
		return -1;
	}

	ret = pthread_create(&tid, &attr, monitor, this);
	if (ret != 0) {
		DERROR("pthread_create error: %s\n", strerror(ret));
		pthread_attr_destroy(&attr);
		return -1;
	}

	pthread_attr_destroy(&attr);
	return 0;
}

int ConnPool::set_memlink_timer()
{
	struct itimerval t;
	t.it_interval.tv_sec = 5;
	t.it_interval.tv_usec = 0;
	t.it_value.tv_sec = 5;
	t.it_value.tv_usec = 0;

	struct sigaction sig;

	sig.sa_flags = 0;
	sigemptyset(&sig.sa_mask);
	sig.sa_handler = handler_sigalrm;

	sigaction(SIGALRM, &sig, NULL);

	setitimer(ITIMER_REAL, &t, NULL);

	return 0;
}

int ConnPool::init(const Config* config)
{
	int i, j, id;
	int ret;

    conf = config;
    
    if (conf == NULL)
    {
        DERROR("input configuration is not valided");
        return ERROR_CODE_INTERNAL;
    }
    
    pool[CONN_POOL_HBASE_INDEX].type = CONN_POOL_TYPE_HBASE;
	pool[CONN_POOL_HDFS_INDEX].type = CONN_POOL_TYPE_HDFS;
	//pool[CONN_POOL_MEMLINK_INDEX].type = CONN_POOL_TYPE_MEMLINK;
    
	for (i = 0; i < CONN_POOL_TYPE_NUM; i++) {
		id = 0;
		pthread_mutex_init(&pool[i].m, NULL);
        
		for (j = 0; j < conf->connect_num; j++) {
			ConnPoolItem * conn = new ConnPoolItem;
			memset(conn, 0, sizeof(ConnPoolItem));
			conn->index = i;
			conn->id = id;
			conn->type = pool[i].type;
			conn->head = pool + i;
			conn->next = pool[i].root;
			pool[i].root = conn;
			id++;
		}
	}
    
	fds = new (struct notify_fds);

	if (pipe(fds->reactive_fd) < 0) {
		DERROR("pipe error: %s\n", strerror(errno));
		return ERROR_CODE_INTERNAL;
	}

	if (pipe(fds->memlink_timer_fd) < 0) {
		DERROR("pipe error: %s\n", strerror(errno));
		return ERROR_CODE_INTERNAL;
	}

	if (set_memlink_timer() < 0) {
		DERROR("conn pool set memlink timer error\n");
		return ERROR_CODE_INTERNAL;
	}

	if (start_connpool_monitor() < 0) {
		DERROR("conn pool monitor start error\n");
		return ERROR_CODE_INTERNAL;
	}

	for (i = 0; i < CONN_POOL_TYPE_NUM; i++) {
		ConnPoolItem *conn = pool[i].root;
		if (pool[i].type == CONN_POOL_TYPE_HBASE) {
			while (conn) {
				hbase_conn *hconn = new hbase_conn(conf->hbase_host, conf->hbase_port);
				ret = hconn->open();
				if (ret == -1) {
					hconn->close();
					delete(hconn);
                    DERROR("open hbase connection error\n");
					return ERROR_CODE_INTERNAL;
				}
				conn->item = hconn;
				hconn->pool = conn;
				conn->active = 1;
				conn->idle = 1;
				conn = conn->next;
			}
		} else if (pool[i].type == CONN_POOL_TYPE_HDFS) {
			while (conn) {
				hdfs_conn *hconn = new hdfs_conn(conf->hdfs_host, conf->hdfs_port);
				ret = hconn->open();
				if (ret == -1) {
					hconn->close();
					delete(hconn);
                    DERROR("open hdfs connection error\n");
					return ERROR_CODE_INTERNAL;
				}
				conn->item = hconn;
				hconn->pool = conn;
				conn->active = 1;
				conn->idle = 1;
				conn = conn->next;
			}
		} else if (pool[i].type == CONN_POOL_TYPE_MEMLINK) {
			/*while (conn) {
				//break;
                memlink_conn *mconn = new memlink_conn(conf->memlink_host, conf->memlink_read_port, conf->memlink_write_port);
				ret = mconn->open();
				if (ret == -1) {
					mconn->close();
					delete(mconn);
                    DERROR("open memlink connection error\n");
					return ERROR_CODE_INTERNAL;
				}
				conn->item = mconn;
				mconn->pool = conn;
				conn->active = 1;
				conn->idle = 1;
				conn = conn->next;
			}
             */
		}
	}
    
    DINFO("connection pool init successed");
    return ERROR_CODE_OK;
}

int notify_reactive(ConnPoolItem *conn)
{
	queue_put(conn);
	if (write(cp->fds->reactive_fd[1], "", 1) < 0)
	{
        DERROR("write error: %s\n", strerror(errno));
    }
	//DERROR("notify\n");
	return 0;
}

void *ConnPool::get(int type)
{
	pthread_mutex_lock(&pool[type].m);
	ConnPoolItem *root = pool[type].root;
	while (root != NULL) {
		if (root->active && root->idle) {
			//printf("conn id %d\n", root->id);
			root->idle = 0;
			break;
		}
		root = root->next;
	}
	pthread_mutex_unlock(&pool[type].m);

	if (root != NULL) {
		return root->item;
	} else {
		return NULL;
	}
}

int ConnPool::put(int type, void *conn)
{
	ConnPoolItem *item;
	if (conn == NULL)
		return 0;
	if (type == CONN_POOL_TYPE_HBASE) {
		item = ((struct hbase_conn *)conn)->pool;
	} else if (type == CONN_POOL_TYPE_HDFS) {
		item = ((struct hdfs_conn *)conn)->pool;
	} else if (type == CONN_POOL_TYPE_MEMLINK) {
		return 0;
        //item = ((struct memlink_conn *)conn)->pool;
	}
	int index = item->index;

	pthread_mutex_lock(&pool[index].m);
	item->idle = 1;
	if (!item->active) {
		item->idle = 0;
		notify_reactive(item);
	}
	pthread_mutex_unlock(&pool[index].m);

	return 0;
}
